package cn.wanghaomiao.dao;

import java.io.IOException;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import cn.wanghaomiao.util.HBaseUtil;

/**
 * 操作hbase基础类
 * 
 * @author 胡宇鹏
 * 
 */
public class HBaseDao {

	private static Admin admin;
	private static Connection connection;
	private static Table table;
	private static Result result;

	/**
	 * 创建表
	 * 
	 * @param tableName
	 *            表名
	 * @param family
	 *            列族列表
	 * @throws IOException
	 */
	public static void createTable(String tableName, String[] cols){
		try{
		admin = HBaseUtil.getAdmin();
		TableName tName = TableName.valueOf(tableName);
		if (admin.tableExists(tName)) {
			System.out.println(tableName + " 表已经存在.");
		} else {
			HTableDescriptor hTableDesc = new HTableDescriptor(tName);
			for (String col : cols) {
				HColumnDescriptor hColumnDesc = new HColumnDescriptor(col);
				hTableDesc.addFamily(hColumnDesc);
			}
			admin.createTable(hTableDesc);
			System.out.println("创建表：" + tableName);
		}
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			HBaseUtil.close();
		}
		
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 *            表名称
	 * @throws IOException
	 */
	public static void deleteTable(String tableName) throws IOException {
		admin = HBaseUtil.getAdmin();
		TableName tName = TableName.valueOf(tableName);
		if (admin.tableExists(tName)) {
			admin.disableTable(tName);
			admin.deleteTable(tName);
		} else {
			System.out.println(tableName + " 表不存在.");
		}
		HBaseUtil.close();
	}

	/**
	 * 查看已有表
	 * 
	 * @throws IOException
	 */
	public static void listTables() {
		admin = HBaseUtil.getAdmin();
		HTableDescriptor hTableDescriptors[] = null;
		try {
			hTableDescriptors = admin.listTables();
		} catch (IOException e) {
			e.printStackTrace();
		}
		for (HTableDescriptor hTableDescriptor : hTableDescriptors) {
			System.out.println(hTableDescriptor.getNameAsString());
		}
		HBaseUtil.close();
	}

	/**
	 * 插入单行
	 * 
	 * @param tableName
	 *            表名称
	 * @param rowKey
	 *            行键
	 * @param colFamily
	 *            列族
	 * @param col
	 *            列(列名不能重复)
	 * @param value
	 *            值
	 * @throws IOException
	 */
	public static void insert(String tableName, String rowKey,
			String colFamily, String col, String value) {
		try {
			admin = HBaseUtil.getAdmin();
			connection = HBaseUtil.getConn();
			table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col),
					Bytes.toBytes(value));
			table.put(put);
			/*
			 * 批量插入 List<Put> puts = new ArrayList<Put>(); puts.add(put);
			 * table.put(puts);
			 */

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				table.close();
				HBaseUtil.close();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

	public static void delete(String tableName, String rowKey,
			String colFamily, String col) throws IOException {
		admin = HBaseUtil.getAdmin();
		connection = HBaseUtil.getConn();
		if (!admin.tableExists(TableName.valueOf(tableName))) {
			System.out.println(tableName + " 表不存在.");
		} else {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete del = new Delete(Bytes.toBytes(rowKey));
			if (colFamily != null) {
				del.addFamily(Bytes.toBytes(colFamily));
			}
			if (colFamily != null && col != null) {
				del.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
			}
			/*
			 * 批量删除 List<Delete> deleteList = new ArrayList<Delete>();
			 * deleteList.add(delete); table.delete(deleteList);
			 */
			table.delete(del);
			table.close();
		}
		HBaseUtil.close();
	}

	/**
	 * 根据RowKey获取数据
	 * 
	 * @param tableName
	 *            表名称
	 * @param rowKey
	 *            RowKey名称
	 * @param colFamily
	 *            列族名称
	 * @param col
	 *            列名称
	 * @throws IOException
	 */
	public static void getData(String tableName, String rowKey,
			String colFamily, String col){
		try{
		admin = HBaseUtil.getAdmin();
		connection = HBaseUtil.getConn();
		table = connection.getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(rowKey));
		if (colFamily != null) {
			get.addFamily(Bytes.toBytes(colFamily));
		}
		if (colFamily != null && col != null) {
			get.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
		}
		Result result = table.get(get);
		showCell(result);
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			try {
				table.close();
				HBaseUtil.close();
			} catch (IOException e) {
				e.printStackTrace();
			}			
		}
	}

	/**
	 * 根据RowKey获取信息
	 * 
	 * @param tableName
	 * @param rowKey
	 * @throws IOException
	 */
	public static void getData(String tableName, String rowKey){
		getData(tableName, rowKey, null, null);
	}

	/**
	 * 格式化输出
	 * 
	 * @param result
	 */
	public static void showCell(Result result) {
		Cell[] cells = result.rawCells();
		for (Cell cell : cells) {
			System.out.println("RowName: "
					+ new String(CellUtil.cloneRow(cell)) + " ");
			System.out.println("Timetamp: " + cell.getTimestamp() + " ");
			System.out.println("column Family: "
					+ new String(CellUtil.cloneFamily(cell)) + " ");
			System.out.println("row Name: "
					+ new String(CellUtil.cloneQualifier(cell)) + " ");
			System.out.println("value: "
					+ new String(CellUtil.cloneValue(cell)) + " ");
		}
	}
	
	/**
	 * 获取cell的值
	 * 
	 * @param result
	 */
	public static Cell[] getCell(String tableName, String rowKey,
			String colFamily, String col) {
		try{
		admin = HBaseUtil.getAdmin();
		connection = HBaseUtil.getConn();
		table = connection.getTable(TableName.valueOf(tableName));
		Get get = new Get(Bytes.toBytes(rowKey));
		if (colFamily != null) {
			get.addFamily(Bytes.toBytes(colFamily));
		}
		if (colFamily != null && col != null) {
			get.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col));
		}
		result = table.get(get);
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			try {
				table.close();
				HBaseUtil.close();
			} catch (IOException e) {
				e.printStackTrace();
			}			
		}
		Cell[] cells = result.rawCells();
		return cells;
	}

}
